package com.hmall.common.mq;


import cn.hutool.core.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;


@Slf4j
@Component
public class RabbitMqHelper {

    @Autowired
    private  RabbitTemplate rabbitTemplate;

    //存储消息重试次数
    private final Map<String, AtomicInteger> retryMap = new ConcurrentHashMap<>();


    /**
     * 消息发送
     * @param exchange
     * @param routingKey
     * @param msg
     */
    public void sendMessage(String exchange, String routingKey, Object msg){
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }

    /**
     * 消息延迟发送
     * @param exchange
     * @param routingKey
     * @param msg
     * @param delay
     */
    public void sendDelayMessage(String exchange, String routingKey, Object msg, long delay){
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, message -> {
            message.getMessageProperties().setDelayLong(delay);
            return message;
        });
    }

    /**
     *消息发送，携带确认机制
     * @param exchange
     * @param routingKey
     * @param msg
     * @param maxRetries
     */
    public void sendMessageWithConfirm(String exchange, String routingKey, Object msg, int maxRetries){
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if(ack){
                //成功
                log.info("消息发送成功,exchange:{},routingKey:{},msg:{}",exchange,routingKey,msg);
                //清除重试次数
                if(correlationData!=null){
                    retryMap.remove(correlationData.getId());
                }
            }else{
                //失败
                log.info("消息发送失败,exchange:{},routingKey:{},msg:{}",exchange,routingKey,msg);
                handleRetry(exchange, routingKey, msg, maxRetries, correlationData);
            }

        });
        //设置返回回调
        rabbitTemplate.setReturnsCallback(returned->{
            log.error("消息无法路由到队列,exchange:{},routingKey:{},msg:{},replyCode:{},replyText:{}",
                    returned.getExchange(),returned.getRoutingKey(),returned.getMessage(),returned.getReplyCode(),returned.getReplyText());
        });
        //发送消息
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(exchange,routingKey,msg,correlationData);
    }


    private void handleRetry(String exchange, String routingKey, Object msg, int maxRetries, CorrelationData correlationData) {
        if (ObjectUtil.isEmpty(correlationData)){
            return;
        }
        String messageId = correlationData.getId();
        AtomicInteger retryCount = retryMap.computeIfAbsent(messageId, k -> new AtomicInteger(0));
        if (retryCount.incrementAndGet() <= maxRetries) {
            log.info("第{}次重试发送消息,exchange:{},routingKey:{},msg:{}", retryCount.get(), exchange, routingKey, msg);
            // 延迟重试，避免立即重试导致的问题
            try {
                Thread.sleep(1000 * retryCount.get()); // 指数退避延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            rabbitTemplate.convertAndSend(exchange, routingKey, msg, correlationData);
        } else {
            log.error("消息发送失败，已达到最大重试次数:{},exchange:{},routingKey:{},msg:{}", maxRetries, exchange, routingKey, msg);
            // 可以在这里添加持久化到数据库等操作，用于后续人工处理
            retryMap.remove(messageId);
        }
    }
}
